package com.lzwk.config;

import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;

import java.io.Serializable;
import java.util.concurrent.TimeUnit;

/**
 * @Author: CC
 * @Date: 2022/2/10 19:24
 */
public class JobConfig implements Serializable {
    public final String earthLogTopic;
    public final String earthLogGroupId;
    public final String earthKafkaServers;
    public final String statLogTopic;
    public final String statLogGroupId;
    public final String statLogNzTopic;
    public final String statLogNzGroupId;
    public final String statLogKafkaServers;
    public final String env;

    public JobConfig(String env, StreamExecutionEnvironment environment) {
        this.env = env;
        if (env.equals("prod")) {
            configProdEnvironment(environment);
            this.earthLogTopic = "nginx-ingress-log";
            this.earthLogGroupId = "earth_log_clean_n";
            this.earthKafkaServers = "10.1.128.74:9092";
            this.statLogTopic = "stat_log_lb";
            this.statLogGroupId = "stat_log_lb_clean_113_n";
            this.statLogNzTopic = "dwd_nezha_stat_log";
            this.statLogNzGroupId = "data_group_stat";
            this.statLogKafkaServers = "10.1.192.30:9092";
        } else if (env.equals("test")) {
            this.earthLogTopic = "ods_base_log";
            this.earthLogGroupId = "earth_log_clean_n";
            this.earthKafkaServers = "127.0.0.1:9092";
            this.statLogTopic = "stat_log_lb";
            this.statLogGroupId = "stat_log_lb_clean_113_n";
            this.statLogNzTopic = "dwd_nezha_stat_log";
            this.statLogNzGroupId = "data_group_stat";
            this.statLogKafkaServers = "127.0.0.1:9092";
        } else {
            this.earthLogTopic = "ods_base_log";
            this.earthLogGroupId = "earth_log_clean_n";
            this.earthKafkaServers = "127.0.0.1:9092";
            this.statLogTopic = "stat_log_lb";
            this.statLogGroupId = "stat_log_lb_clean_113_n";
            this.statLogNzTopic = "dwd_nezha_stat_log";
            this.statLogNzGroupId = "data_group_stat";
            this.statLogKafkaServers = "127.0.0.1:9092";
        }
    }

    private void configProdEnvironment(StreamExecutionEnvironment environment) {
        environment.setParallelism(4);
        environment.setRestartStrategy(RestartStrategies.failureRateRestart(
                3,
                Time.of(5, TimeUnit.MINUTES),
                Time.of(30, TimeUnit.SECONDS)));
        environment.setStateBackend(new EmbeddedRocksDBStateBackend());
        environment.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE);
        CheckpointConfig checkpointConfig = environment.getCheckpointConfig();
        checkpointConfig.setCheckpointStorage("hdfs://10.2.16.4:4007/flink_113/checkpoints/");
        checkpointConfig.setCheckpointTimeout(60000);
        checkpointConfig.setMinPauseBetweenCheckpoints(1000);
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        checkpointConfig.setTolerableCheckpointFailureNumber(10);
    }
}
